package com.navinfo.platform.dataanalysis.service.impl;

import com.google.common.collect.Lists;
import com.navinfo.platform.common.service.ICommonService;
import com.navinfo.platform.common.service.IDataHandleService;
import com.navinfo.platform.common.utils.DataUtil;
import com.navinfo.platform.common.utils.DateUtils;
import com.navinfo.platform.common.utils.StatisticsDef;
import com.navinfo.platform.common.utils.TidPartitioner;
import com.navinfo.platform.dataanalysis.dto.AlarmInfoDto;
import com.navinfo.platform.dataanalysis.dto.AlarmInfoDto;
import com.navinfo.platform.dataanalysis.dto.EventInfoDto;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.sql.Timestamp;
import java.util.*;
import java.util.stream.StreamSupport;

/**
 * 根据告警日统计信息，生成周或月统计信息
 * @author  web 20181203
 */
public class AlarmDataHandleService implements IDataHandleService {

    @Override
    public void handle(List<Object> obj, Map<String, String> configMap, SparkSession spark) {
        String day = configMap.get("args_1");
        String type = configMap.get("args_2");
        String path = configMap.get(StatisticsDef.WEEK.equals(type) ? "hive.alarm.week.path" :  "hive.alarm.month.path");

        Tuple2<String, String> startEndDay = DateUtils.getStartEndDay(day, type);
        path = String.format(path, startEndDay._1);
        System.out.println("进行告警统计信息处理，partDay="+ startEndDay._1+"，path="+ path);

        //RDD转换
        JavaPairRDD<Long, List<AlarmInfoDto>> convertRDD = convert((Dataset<Row>) obj.get(0), configMap);

        //计算周或月的统计项
        JavaRDD<AlarmInfoDto> alarmInfoRDD = statisticsData(convertRDD, startEndDay._1, spark);

        //保存数据
        long start = System.currentTimeMillis();
        System.out.println("进行告警统计信息处理，保存Parquet文件开始...");
        Dataset<Row> dataFrame = spark.createDataFrame(alarmInfoRDD, AlarmInfoDto.class);
//        dataFrame.select(AlarmInfoDto.columns()).show();
        dataFrame.select(AlarmInfoDto.columns()).repartition(1).write().mode(SaveMode.Overwrite).parquet(path);
        System.out.println("进行告警统计信息处理，保存Parquet文件耗时"+ (System.currentTimeMillis()-start));
    }

    /**
     * 将alarm_info_day数据转成聚合后的{@code <tid,List<AlarmInfoDto>>}
     * @param alarmDataset
     * @param configMap
     */
    public JavaPairRDD<Long, List<AlarmInfoDto>> convert(Dataset<Row> alarmDataset, Map<String, String> configMap) {
        //将Row转换为dto
        JavaPairRDD<Long, AlarmInfoDto> javaPairRDD = alarmDataset.toJavaRDD().mapPartitionsToPair((PairFlatMapFunction<Iterator<Row>, Long, AlarmInfoDto>) iterator -> {
            return StreamSupport.stream(((Iterable<Row>) () -> iterator).spliterator(), false).map(row -> {
                AlarmInfoDto alarmInfoDto = AlarmInfoDto.parseRow(row);
                return new Tuple2<>(alarmInfoDto.getTid(), alarmInfoDto);
            }).iterator();
        });

        //重分区+聚合操作
        int partitionNum = Optional.ofNullable(configMap.get("args_3")).map(u -> Integer.parseInt(u)).orElse(1024);
        return javaPairRDD.partitionBy(new TidPartitioner(partitionNum))
                    .aggregateByKey(
                        Lists.<AlarmInfoDto>newArrayList(),
                        (v1, v2) -> {
                            v1.add(v2);
                            return v1;
                        }, (v1, v2) -> {
                            v1.addAll(v2);
                            return v1;
                        });
    }

    /**
     * 生成周或日统计的告警信息列表
     * @param convertRDD
     * @return
     */
    private JavaRDD<AlarmInfoDto> statisticsData(JavaPairRDD<Long, List<AlarmInfoDto>> convertRDD, String day , SparkSession spark) {
        JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
        final Broadcast<String> partDay = jsc.broadcast(day);

        return convertRDD.mapPartitions(iterator -> {
            return StreamSupport.stream(((Iterable<Tuple2<Long, List<AlarmInfoDto>>>) () -> iterator).spliterator(), false).map(item -> {
                AlarmInfoDto alarmInfo = new AlarmInfoDto();
                alarmInfo.setTid(item._1);

                item._2.forEach(e -> {
                    alarmInfo.setUid(e.getUid());
                    alarmInfo.setMileage(alarmInfo.getMileage() + e.getMileage());
                    alarmInfo.setDuration(alarmInfo.getDuration() + e.getDuration());

                    alarmInfo.setSo90Times(alarmInfo.getSo90Times() + e.getSo90Times());
                    alarmInfo.setSo90Mileage(alarmInfo.getSo90Mileage() + e.getSo90Mileage());
                    alarmInfo.setSo90Duration(alarmInfo.getSo90Duration() + e.getSo90Duration());

                    alarmInfo.setSo120Times(alarmInfo.getSo120Times() + e.getSo120Times());
                    alarmInfo.setSo120Mileage(alarmInfo.getSo120Mileage() + e.getSo120Mileage());
                    alarmInfo.setSo120Duration(alarmInfo.getSo120Duration() + e.getSo120Duration());

                    alarmInfo.setSrTimes(alarmInfo.getSrTimes() + e.getSrTimes());
                    alarmInfo.setSrMileage(alarmInfo.getSrMileage() + e.getSrMileage());
                    alarmInfo.setSrDuration(alarmInfo.getSrDuration() + e.getSrDuration());

                    alarmInfo.setLrogTimes(alarmInfo.getLrogTimes() + e.getLrogTimes());
                    alarmInfo.setLrogMileage(alarmInfo.getLrogMileage() + e.getLrogMileage());
                    alarmInfo.setLrogDuration(alarmInfo.getLrogDuration() + e.getLrogDuration());
                });
                int mileage = alarmInfo.getMileage();
                int duration = alarmInfo.getDuration();

                //设置占比
                alarmInfo.setSo90100kmTimes(DataUtil.hundredKmTimes(alarmInfo.getSo90Times(), mileage));
                alarmInfo.setSo90MileageProportion(DataUtil.mileageProportion(alarmInfo.getSo90Mileage(), mileage));
                alarmInfo.setSo90DurationProportion(DataUtil.durationProportion(alarmInfo.getSo90Duration(), duration));
                alarmInfo.setSo120100kmTimes(DataUtil.hundredKmTimes(alarmInfo.getSo120Times(), mileage));
                alarmInfo.setSo120MileageProportion(DataUtil.mileageProportion(alarmInfo.getSo120Mileage(), mileage));
                alarmInfo.setSo120DurationProportion(DataUtil.durationProportion(alarmInfo.getSo120Duration(), duration));
                alarmInfo.setSr100kmTimes(DataUtil.hundredKmTimes(alarmInfo.getSrTimes(), mileage));
                alarmInfo.setSrMileageProportion(DataUtil.mileageProportion(alarmInfo.getSrMileage(), mileage));
                alarmInfo.setSrDurationProportion(DataUtil.durationProportion(alarmInfo.getSrDuration(), duration));
                alarmInfo.setLrog100kmTimes(DataUtil.hundredKmTimes(alarmInfo.getLrogTimes(), mileage));
                alarmInfo.setLrogMileageProportion(DataUtil.mileageProportion(alarmInfo.getLrogMileage(), mileage));
                alarmInfo.setLrogDurationProportion(DataUtil.durationProportion(alarmInfo.getLrogDuration(), duration));
                alarmInfo.setUpdateTime(new Timestamp(System.currentTimeMillis()));
                alarmInfo.setDay(partDay.value());

                return alarmInfo;
            }).iterator();
        });
    }
}
